package com.spark.zhou.demo.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrameReader;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;

/**
 * @Description: SparkSql连接操作Mysql
 * @Author: ZhOu
 * @Date: 2018/5/24
 */
public class Spark2MySql {
    private static final String URL = "jdbc:mysql://114.55.119.15:3306/ad_sq_test";
    private static final String USER = "root";
    private static final String PWD = "Qaz327389597";

    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("spark2Mysql")
                .setMaster("spark://node1:7077");
        JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);

        SQLContext sqlContext = SQLContext.getOrCreate(sparkContext.sc());

        DataFrameReader reader = sqlContext.read().format("jdbc");

        reader.option("url", URL);
        reader.option("dbtable", "ad_test");
        reader.option("driver", "com.mysql.jdbc.Driver");
        reader.option("user", USER);
        reader.option("password", PWD);

        Dataset<Row> dataRow = reader.load();
        dataRow.show();

        List<String> list = new ArrayList<>();
        list.add("{\"device_name\":\"spark\",\"amount\":100}");

        JavaRDD<String> rdd = sparkContext.parallelize(list);
        Dataset<Row> dataRow1 = sqlContext.read().json(rdd);
        dataRow1.show();
        long s1 = System.currentTimeMillis();
        dataRow1.javaRDD().foreachPartition(rowIterator -> {
            Connection conn;
            PreparedStatement ps;

            String sql = "insert into ad_test(device_name,amount) values(?,?)";
            conn = DriverManager.getConnection(URL, USER, PWD);
//            conn.setAutoCommit(false);
//            ps = conn.prepareStatement(sql);
//            int count = 0;
//            try {
//                while (rowIterator.hasNext()) {
//                    count++;
//                    Row row = rowIterator.next();
//                    ps.setString(1, row.getAs("device_name"));
//                    ps.setLong(2, row.getAs("amount"));
//
//                    ps.addBatch();
//                    if (count % 100 == 0) {
//                        ps.executeBatch();
//                    }
//                }
//                ps.executeBatch();
//                conn.commit();
//            } finally {
//                if (ps != null) {
//                    ps.close();
//                }
//
//                if (conn != null) {
//                    conn.close();
//                }
//            }

            sql = "SELECT sdk_version AS sdk_version , SUM(increase_count) AS increase_count,SUM(active_count) AS active_count," +
                    "SUM(open_count) AS open_count,SUM(show_count) AS show_count FROM ad_app_record_classify_new_tmp2 " +
                    " WHERE  create_date BETWEEN 20170410 AND 20170416 GROUP BY sdk_version ORDER BY sdk_version";
            ps = conn.prepareStatement(sql);

            ResultSet rs = ps.executeQuery();

            System.out.println(rs.getFetchSize());
            while (rs.next()) {
                System.out.println(
                        rs.getInt(1) + "\t" +
                                rs.getInt(2) + "\t" +
                                rs.getInt(3) + "\t" +
                                rs.getInt(4) + "\t" +
                                rs.getInt(5));
            }

            if (ps != null) {
                ps.close();
            }

            if (conn != null) {
                conn.close();
            }
        });

        long s2 = System.currentTimeMillis();
        System.out.println("\n\n耗时：" + (s2 - s1));
    }
}
